package com.shujia.homework;

import com.shujia.util.HBaseTool;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

/*
    查询文科一班学生总分排名前10的学生（输出：学号，姓名，班级，总分）结果写到hbase

    1、数据准备：students   score
    2、在hbase上建表 students和scores
    3、先从students表中查询文科一班的学生，学号，姓名，班级
    4、根据获取到的学号查询scores表中所有的文科一班学生成绩
    5、根据查询到学生，相同学号为一组，算总分
    6、与第三步查询到结果进行连接
    7、将结果写入到hbase表中


    重点：
        1、熟练掌握hbase基础api的使用，建表，插入数据，获取表，查询数据
        2、数量掌握过滤器的使用，例如：单列值过滤器
        3、学会合理地设计rowkey，防止数据去重覆盖，以及方便我们做查询。
        4、默认已经掌握：java基础，集合，IO流，学会合理的选择对应集合来解决问题
 */
public class HBaseTest {
    public static void main(String[] args) {
        //1、数据准备：students   score
        //2、创建scores表
//        HBaseTool.createOneTable("scores","info");
        //将成绩表数据插入到scores表中
//        insertScoresData();

//        //3、先从students表中查询文科一班的学生，学号，姓名，班级
//        HashMap<String, String> studentsMap = selectFromStudentsWithClazz("文科一班");
//
//
//        //4、根据获取到的学号查询scores表中所有的文科一班学生成绩
//        TreeMap<Student, String> stringTreeMap = selectFromStudentsWithId(studentsMap);
//
//        // 5、6、与第三步查询到结果进行连接
//        ArrayList<Student> students = selectStudents(studentsMap, stringTreeMap);
//
//        // 7、将结果写入到hbase表中
//        insertStudentTopN(students);

        //查询结果表
        selectTopNStudent();

    }

    public static void selectTopNStudent(){
        try {
            //获取该表对象
            Table studentTop10WithClazz = HBaseTool.getOneTable("studentTop10WithClazz");
            Scan scan = new Scan();

            ResultScanner resultScanner = studentTop10WithClazz.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);


        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void insertStudentTopN(ArrayList<Student> students) {
        try {
            /**
             *  372_1500100308 '黄初夏' '文科一班' 628
             *  405_1500100875 '马向南' '文科一班' 595
             */
            // 现在hbase中创建一张结果表
            Admin hbaseAdmin = HBaseTool.HBASE_ADMIN;
            if (!hbaseAdmin.tableExists(TableName.valueOf("studentTop10WithClazz"))) {
                //如果目标表不存在的话，创建一下
                HBaseTool.createOneTable("studentTop10WithClazz","info");
            }

            //获取该表对象
            Table studentTop10WithClazz = HBaseTool.getOneTable("studentTop10WithClazz");

            ArrayList<Put> puts = new ArrayList<>();
            //取出前10条数据
            for (int i = 0; i < 10; i++) {
                Student student = students.get(i);
                //构建rowKey
                byte[] rk = Bytes.toBytes((1000L - student.getSumScore()) + "_" + student.getId());

                Put put = new Put(rk);
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes(student.getName()));
                puts.add(put);

                Put put2 = new Put(rk);
                put2.addColumn(Bytes.toBytes("info"),Bytes.toBytes("clazz"),Bytes.toBytes(student.getClazz()));
                puts.add(put2);

                Put put3 = new Put(rk);
                put3.addColumn(Bytes.toBytes("info"),Bytes.toBytes("sumScore"),Bytes.toBytes(String.valueOf(student.getSumScore())));
                puts.add(put3);
            }

            studentTop10WithClazz.put(puts);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static ArrayList<Student> selectStudents(HashMap<String, String> studentsMap, TreeMap<Student, String> stringTreeMap) {
        ArrayList<Student> studentsList = new ArrayList<>();

        try {
            stringTreeMap.forEach((stu, v) -> {
                //从studentsMap将根据学号该学生的姓名和班级取出
                String id = stu.getId();
                String name = studentsMap.get(id).split("_")[0];
                String clazz = studentsMap.get(id).split("_")[1];
                stu.setName(name);
                stu.setClazz(clazz);
                studentsList.add(stu);
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
        return studentsList;
    }


    public static TreeMap<Student, String> selectFromStudentsWithId(HashMap<String, String> studentsMap) {
        //再定义一个TreeMap集合，将studentWithScore输入导入到该集合中
        TreeMap<Student, String> map = new TreeMap<>(new Comparator<Student>() {
            @Override
            public int compare(Student o1, Student o2) {
                return (int) (o2.getSumScore() - o1.getSumScore());
            }
        });


        try {
            //获取成绩表
            Table scores = HBaseTool.getOneTable("scores");
            // 遍历上一步传进来的集合，得到该班级所有学生的学号
            Set<String> ids = studentsMap.keySet();
            FilterList filterList = new FilterList();
            RowFilter rowFilter = null;

            //创建Map集合存储学号和总成绩
            HashMap<String, Long> studentWithScore = new HashMap<>();

            //遍历学号,将每个学号封装成一个过滤器
            for (String id : ids) { // 遍历每一个学生
                rowFilter = new RowFilter(CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes(id)));
                Scan scan = new Scan();
                scan.setFilter(rowFilter);
                ResultScanner resultScanner = scores.getScanner(scan); // 一个学生的6条分数数据
                for (Result next : resultScanner) {
                    String sid = Bytes.toString(next.getRow()).split("_")[0];
                    long score = Long.parseLong(Bytes.toString(next.getValue(Bytes.toBytes("info"), Bytes.toBytes("score"))));
                    if (studentWithScore.containsKey(sid)) {
                        //如果集合中存在该学号，将该键上一次的value值与这一次的成绩加起来重复赋值
                        studentWithScore.put(sid, studentWithScore.get(sid) + score);
                    } else {
                        studentWithScore.put(sid, score);
                    }
                }
            }


            //遍历总分集合，将数据封装成Student对象插入到map集合中
            studentWithScore.forEach((k, v) -> {
                Student s = new Student(k, v);
                map.put(s, null);
            });


        } catch (Exception e) {
            e.printStackTrace();
        }

        return map;
    }


    public static HashMap<String, String> selectFromStudentsWithClazz(String clazz) {
        //Map<String, String>
        HashMap<String, String> studentsInfo = new HashMap<>();
        try {
            //获取学生表
            Table students = HBaseTool.getOneTable("students");
            //创建单列值过滤器
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                    Bytes.toBytes("info"),
                    Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL,
                    new BinaryComparator(Bytes.toBytes(clazz))
            );

            Scan scan = new Scan();
            scan.setFilter(singleColumnValueFilter);

            ResultScanner resultScanner = students.getScanner(scan);

            for (Result oneLineData : resultScanner) {
                //获取学号
                String id = Bytes.toString(oneLineData.getRow());

                //获取姓名这一列
                String name = Bytes.toString(oneLineData.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));

                //获取班级这一列
                String clazz2 = Bytes.toString(oneLineData.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));

                //学号作为键，姓名和班级作为值
                studentsInfo.put(id, name + "_" + clazz2);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

        return studentsInfo;
    }

    public static void insertScoresData() {
        try {
            //获取一张表
            Table scores = HBaseTool.getOneTable("scores");
            //读取本地数据
            //创建字符缓冲输入流对象读取本地数据
            BufferedReader br = new BufferedReader(new FileReader("hbase/data/score.txt"));
            String line = null;
            while ((line = br.readLine()) != null) {
                //1500100001,1000001,98
                String[] info = line.split(",");
                byte[] rk = Bytes.toBytes(info[0] + "_" + info[1]);
                Put put = new Put(rk);
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("score"), Bytes.toBytes(info[2]));
                scores.put(put);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
